package com.hadoop.mapreduce.FriendRelation;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.Arrays;

/**
 * 第二步，列出有共同好友的人
 */
public class FriendRelation2 {
    static class FriendRelation2Mapper extends Mapper<IntWritable,Text,Text,Text>{
        @Override
        protected void map(IntWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] split = value.toString().split(":");
            String[] friends = split[1].split(",");
            Arrays.sort(friends);
            for (int i = 0; i < friends.length; i++) {
                for (int j = i+1; j < friends.length; j++) {
                    // 发出 <人-人，好友> ，这样，相同的“人-人”对的所有好友就会到同1个reduce中去
                    context.write(new Text(friends[i]+"-"+friends[j]),new Text(split[0]));
                }
            }
        }
    }
    static class FriendRelation2Reducer extends Reducer<Text,Text,Text,Text>{
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            StringBuilder sb=new StringBuilder();
            for (Text text :
                    values) {
               sb.append(text.toString()+",");
            }

            context.write(key,new Text(sb.toString()));
        }
    }

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf);
        job.setJarByClass(FriendRelation2.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        job.setMapperClass(FriendRelation2Mapper.class);
        job.setReducerClass(FriendRelation2Reducer.class);

        FileInputFormat.setInputPaths(job, new Path("D:/temp/out/part-r-00000"));
        FileOutputFormat.setOutputPath(job, new Path("D:/temp/out2"));

        job.waitForCompletion(true);

    }
}
